package com.shujia.rec.compute

import com.shujia.rec.common.Constants
import com.shujia.rec.entry.CaseClass
import com.shujia.rec.funaction.ItemCfSimilarityFunaction
import com.shujia.rec.util.{KafkaUtil, LogUtil}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010

object ItemCfSimilarity {
  def main(args: Array[String]): Unit = {

    /**
      * 商品相似度计算
      *
      */


    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    //获取kafka source
    val kafkaSource: FlinkKafkaConsumer010[String] = KafkaUtil.getKafaSoure("ItemCfHistory", Constants.KAFKA_TOPIC)

    //读所有数据
    kafkaSource.setStartFromLatest()

    val logDS: DataStream[String] = env.addSource(kafkaSource)

    //过滤脏数据
    val filterDS: DataStream[String] = logDS.filter(log => LogUtil.verifyLog(log))

    //将每一行数据转换成实体类
    val entryDS: DataStream[CaseClass.LogEntry] = filterDS.map(log => LogUtil.toEntry(log))


    //计算相关性
    entryDS.flatMap(new ItemCfSimilarityFunaction)


    env.execute("ItemCfSimilarity")

  }
}
